package cn.wjee.gradle.flink.config;

import org.apache.flink.client.deployment.StandaloneClusterId;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.SecurityOptions;

/**
 * Flink Client Builder
 *
 * @author listening
 */
public class FlinkClusterClientBuilder {

    private final Configuration config;

    private FlinkClusterClientBuilder() {
        config = new Configuration();
    }

    public static FlinkClusterClientBuilder builder() {
        return new FlinkClusterClientBuilder();
    }

    public FlinkClusterClientBuilder defaultSsl() {
        enableSslInternal(false).enableSslRest(false);
        return this;
    }

    public FlinkClusterClientBuilder enableSslInternal(boolean enableSslInternal) {
        config.setBoolean(SecurityOptions.SSL_INTERNAL_ENABLED, enableSslInternal);
        return this;
    }

    public FlinkClusterClientBuilder enableSslRest(boolean enableSslRest) {
        config.setBoolean(SecurityOptions.SSL_REST_ENABLED, enableSslRest);
        return this;
    }

    public FlinkClusterClientBuilder addStringProperty(String key, String value) {
        config.setString(key, value);
        return this;
    }


    public FlinkClusterClientBuilder addBooleanProperty(String key, boolean value) {
        config.setBoolean(key, value);
        return this;
    }

    public FlinkClusterClientBuilder addIntegerProperty(String key, Integer value) {
        config.setInteger(key, value);
        return this;
    }

    public FlinkClusterClientBuilder address(String host, Integer restPort, Integer rpcPort) {
        config.setString(JobManagerOptions.ADDRESS, host);
        config.setInteger(JobManagerOptions.PORT, rpcPort);
        config.setString(RestOptions.ADDRESS, host);
        config.setInteger(RestOptions.PORT, restPort);
        return this;
    }

    public RestClusterClient<StandaloneClusterId> build()  {
        try {
            return new RestClusterClient<>(config, StandaloneClusterId.getInstance());
        } catch (Exception e) {
            return null;
        }
    }
}
